package MapReduce.Demo12_ProductAndOrder.reduceJoin;

import MapReduce.writableBean.ProductAndOrderWritable;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import utils.JobSubmit;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @Author lixinlei
 * @Date 2023/3/23 8:29
 */
public class ProductAndOrderReduceJoinApp {

    public static class PAORJMapper extends Mapper<LongWritable, Text, ProductAndOrderWritable,NullWritable>{

        ProductAndOrderWritable outKey = new ProductAndOrderWritable();
        NullWritable outValue = NullWritable.get();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, ProductAndOrderWritable, NullWritable>.Context context) throws IOException, InterruptedException {

            String line = value.toString();
            String[] fields = line.split("\t");

            //如果拆分之后是两个字段，当前数据是商品数据
            if(fields!=null && fields.length==2){
                outKey.setProductId(fields[0]);
                outKey.setProductName(fields[1]);
                outKey.setFlag("product");
            }else{
                //如果拆分之后是三个字段，当前数据是订单数据
                outKey.setOrderId(fields[0]);
                outKey.setProductId(fields[1]);
                outKey.setCount(Integer.parseInt(fields[2]));
                outKey.setFlag("order");
            }

            /**
             * 当前输出的数据格式
             *      1001    01  10  order
             *      1002    01  5   order
             *              01  辣条  product
             *      1001    02  6   order
             *      。。。。
             */
            context.write(outKey,outValue);

        }
    }

    public static class PAORJReducer extends Reducer<ProductAndOrderWritable,NullWritable,ProductAndOrderWritable,NullWritable>{

        /**
         * 以productId为分组条件，进入到reduce端的数据格式
         * 商品数据只能有一条，订单数据可以有多条，且顺序不一定
         *
         *      1001    01  10  order   NullWritable
         *      1002    01  5   order   NullWritable
         *              01  辣条  product NullWritable
         *      1003    01  6   order   NullWritable
         *
         * 进入reduce的格式
         *
         *  key：1003    01  6   order
         *  value：Iterable(NullWritable,NullWritable,NullWritable,NullWritable)
         *
         * @param key
         * @param values
         * @param context
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        protected void reduce(ProductAndOrderWritable key, Iterable<NullWritable> values, Reducer<ProductAndOrderWritable, NullWritable, ProductAndOrderWritable, NullWritable>.Context context) throws IOException, InterruptedException {

            //初始化当前组数据的商品名称
            String productName = "";

            //初始化订单数据的列表
            List<ProductAndOrderWritable> orderList = new ArrayList<>();

            //遍历多个NullWritable本身没有意义，但是需要通过NullWritable这个value获取相应的key的内容
            for (NullWritable value : values) {

                //获取当前数据的标识，确定是订单数据还是商品数据
                String flag = key.getFlag();

                //如果当前数据是商品数据，把商品名提取出来，方便后面使用
                if(flag.equals("product")){
                    productName = key.getProductName();
                }else if (flag.equals("order")) {
                    try {
                        //保证订单对象准确无误，必须使用对象中的属性复制
                        ProductAndOrderWritable order = new ProductAndOrderWritable();
                        BeanUtils.copyProperties(order,key);
                        //如果当前数据是订单数据，把订单数据加入到集合当中
                        orderList.add(order);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }

            }

            //再一次遍历当前组的所有订单数据，在遍历的过程中，设置商品名字段，并直接写出
            for (ProductAndOrderWritable order : orderList) {
                order.setProductName(productName);
                context.write(order,NullWritable.get());
            }

        }
    }

    public static void main(String[] args) {
        JobSubmit.submitBaseJob(
                ProductAndOrderReduceJoinApp.class,
                args,
                "group",
                ProductAndOrderReducerJoinGroupingComparator.class
        );
    }

}
